package com.light.gulimall.seckill.service.impl;

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.light.common.to.SecKillRabbitTO;
import com.light.common.vo.MemberRespVO;
import com.light.gulimall.seckill.feign.CouponRemoteServiceClient;
import com.light.gulimall.seckill.feign.ProductRemoteServiceClient;
import com.light.gulimall.seckill.interceptor.LoginInterceptor;
import com.light.gulimall.seckill.pojo.SeckillSessionPojo;
import com.light.gulimall.seckill.pojo.SeckillSkuRelationEntity;
import com.light.gulimall.seckill.pojo.SkuInfoPojo;
import com.light.gulimall.seckill.service.SecKillSkuService;
import com.light.gulimall.seckill.to.SecKillSkuRedisTO;
import com.sun.org.apache.bcel.internal.generic.NEW;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
public class SecKillSkuServiceImpl implements SecKillSkuService {

    @Autowired
    private CouponRemoteServiceClient couponRemoteServiceClient;

    @Autowired
    private ProductRemoteServiceClient productRemoteServiceClient;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    private final String SESSION_CACHE_PREFIX = "seckill:sessions:";

    private final String SKUKILL_CACHE_PREFIX = "seckill:skus:";

    private final String SKU_STOCK_SEMAPHORE = "seckill:stock:";

    @Autowired
    private RabbitTemplate rabbitTemplate;


    @Override
    public void uploadSecKillSkuInfo() {
        List<SeckillSessionPojo> lasted3SeckillInfo = couponRemoteServiceClient.getLasted3SeckillInfo();

        if (!CollectionUtils.isEmpty(lasted3SeckillInfo)) {
            //将活动信息进行缓存 key:前缀：开始事件_结束时间 value SkuId
            this.saveSessionInfos(lasted3SeckillInfo);

            //保存商品信息 key:前缀 value 商品信息
            this.saveSessionSkuInfos(lasted3SeckillInfo);
        }
    }

    /**
     * 创建流控规则方式二
     * @param ex
     * @return
     */
    public List<SecKillSkuRedisTO> CurrentSeckillSkusHandler(BlockException ex) {
        log.info("{}:流控规则生效","getCurrentSeckillSkus");
        return null;
    }

    @Override
    @SentinelResource(value = "getCurrentSeckillSkus",blockHandler = "CurrentSeckillSkusHandler")
    public List<SecKillSkuRedisTO> getCurrentSeckillSkus() {
        Set<String> keys = stringRedisTemplate.keys(SESSION_CACHE_PREFIX + "*");
        long time = new Date().getTime();
        if (!CollectionUtils.isEmpty(keys)) {
            for (String key : keys) {
                //对key进行处理，得到开始事件和结束时间
                String replace = key.replace(SESSION_CACHE_PREFIX, "");
                String[] keyArr = replace.split("_");
                long startTime = Long.parseLong(keyArr[0]);
                long endTime = Long.parseLong(keyArr[1]);
                //得到该秒杀场次的商品信息
                if (time >= startTime && time <= endTime) {
                    //根据该场次的key，取出value:
                    //1_14
                    //1_13
                    List<String> range = stringRedisTemplate.opsForList().range(key, -100, 100);
                    if (!CollectionUtils.isEmpty(range)) {
                        //根据value查询hash结构，取出具体的信息
                        BoundHashOperations<String, String, String> operations = stringRedisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
                        List<String> values = operations.multiGet(range);
                        if (!CollectionUtils.isEmpty(values)) {
                            List<SecKillSkuRedisTO> secKillSkuRedisTOS = values.stream().map(s -> {
                                return JSON.parseObject(s, SecKillSkuRedisTO.class);
                            }).collect(Collectors.toList());
                            return secKillSkuRedisTOS;
                        }
                    }
                }
            }
        }
        return null;
    }

    @Override
    public SecKillSkuRedisTO getCurrentSeckillSkuById(Long skuId) {
        //定义受保护的资源方式一
        try(Entry entry = SphU.entry("getCurrentSeckillSkuById")) {
            BoundHashOperations<String, String, String> operations = stringRedisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
            //这一步拿到的：k:1_14 v:json格式的SecKillSkuRedisTO
            Set<String> keys = operations.keys();
            if (!CollectionUtils.isEmpty(keys)) {
                //遍历这些key，找到和传递参数相等的skuID的value，转换回java对象并且返回
                for (String key : keys) {
                    long skuIdRedis = Long.parseLong(key.split("_")[1]);
                    if (skuIdRedis == skuId) {
                        //根据key获取值
                        String resultStr = operations.get(key);
                        return JSON.parseObject(resultStr, SecKillSkuRedisTO.class);
                    }
                }
            }
        } catch (BlockException e) {
            log.info("{}:流控规则生效","getCurrentSeckillSkuById");
        }
        return null;
    }

    @Override
    public String kill(String killId, String key, Integer num) {
        MemberRespVO memberRespVO = LoginInterceptor.threadLocal.get();
        Long userId = memberRespVO.getId();
        String timeId = IdWorker.getTimeId();
        //首先校验用户是否登录（在拦截器中已经实现）
        //校验信息是否合法
        BoundHashOperations<String, Object, Object> operations = stringRedisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
        //获取killId的场次信息
        String json = (String) operations.get(killId);
        if (!StringUtils.isBlank(json)){
            SecKillSkuRedisTO secKillSkuRedisTO = JSON.parseObject(json, SecKillSkuRedisTO.class);
            Long startTime = secKillSkuRedisTO.getStartTime();
            Long endTime = secKillSkuRedisTO.getEndTime();
            long time = new Date().getTime();
            //校验时间
            if (startTime > time || endTime < time) {
                return null;
            }
            String randomCode = secKillSkuRedisTO.getRandomCode();
            Long promotionSessionId = secKillSkuRedisTO.getPromotionSessionId();
            Long skuId = secKillSkuRedisTO.getSkuId();
            //校验参数中的随机码和场次_skuId与redis中的是否一致
            if (!key.equals(randomCode) || !killId.equals(promotionSessionId+"_"+skuId)) {
                return null;
            }
            //校验该用户是否已经秒杀过
            String userKey = new StringBuffer().append(userId).append("_").append(promotionSessionId).append("_").append(skuId).toString();
            //setIfAbsent 只有不存在才会创建
            Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(userKey, num.toString(), 100, TimeUnit.MILLISECONDS);
            if (!aBoolean) {
                return null;
            }
            //扣减库存
            RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + randomCode);
            try {
                //利用有超时时间的获取，超过时间获取不到就自己放弃，不会死等
                boolean b = semaphore.tryAcquire(num, 100, TimeUnit.MILLISECONDS);
                if (!b){
                    return null;
                }
                //向rabbitMQ发消息，创建订单
                SecKillRabbitTO secKillRabbitTO = new SecKillRabbitTO();
                secKillRabbitTO.setMemberId(userId);
                secKillRabbitTO.setNum(num);
                secKillRabbitTO.setPromotionSessionId(promotionSessionId);
                secKillRabbitTO.setSkuId(skuId);
                secKillRabbitTO.setOrderNo(timeId);
                rabbitTemplate.convertAndSend("order-event-exchange","order.seckill.order",secKillRabbitTO);
            } catch (InterruptedException e) {
                return null;
            }
        }
        return timeId;
    }

    private void saveSessionSkuInfos(List<SeckillSessionPojo> lasted3SeckillInfo) {

        lasted3SeckillInfo.forEach(seckillSessionPojo -> {
            //准备hash操作，一个活动场次一个hash操作
            BoundHashOperations<String, Object, Object> operations = stringRedisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
            List<SeckillSkuRelationEntity> relations = seckillSessionPojo.getSkuRelationEntities();
            relations.forEach(relation -> {
                String token = UUID.randomUUID().toString().replace("-", "");
                if (Boolean.FALSE.equals(operations.hasKey(relation.getPromotionSessionId().toString()+"_"+relation.getSkuId().toString()))) {
                    //缓存商品，一个活动场次对应的具体商品
                    SecKillSkuRedisTO secKillSkuRedisTO = new SecKillSkuRedisTO();
                    BeanUtils.copyProperties(relation, secKillSkuRedisTO);
                    //还应该设置商品详细信息 远程调用product服务
                    SkuInfoPojo skuInfo = null;
                    try {
                        skuInfo = productRemoteServiceClient.getSkuInfo(relation.getSkuId());
                    } catch (Exception e) {
                        log.info("根据skuId：{}查询商品服务错误：", relation.getSkuId(), e);
                    }
                    secKillSkuRedisTO.setSkuInfo(skuInfo);
                    //设置商品的开始事件和结束时间
                    secKillSkuRedisTO.setStartTime(seckillSessionPojo.getStartTime().getTime());
                    secKillSkuRedisTO.setEndTime(seckillSessionPojo.getEndTime().getTime());
                    //设置随机码
                    secKillSkuRedisTO.setRandomCode(token);
                    String result = JSON.toJSONString(secKillSkuRedisTO);
                    operations.put(relation.getPromotionSessionId().toString()+"_"+relation.getSkuId().toString(), result);


                    //限流 相比较于固定的skuId，每次的随机码都不一样
                    RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + token);
                    //商品可以秒杀的总量作为信号量
                    semaphore.trySetPermits(relation.getSeckillCount().intValue());

                }
            });
        });
    }

    private void saveSessionInfos(List<SeckillSessionPojo> lasted3SeckillInfo) {
        lasted3SeckillInfo.forEach(seckillSessionPojo -> {
            long start = seckillSessionPojo.getStartTime().getTime();
            long end = seckillSessionPojo.getEndTime().getTime();
            String key = SESSION_CACHE_PREFIX + start + "_" + end;
            Boolean hasKey = stringRedisTemplate.hasKey(key);
            if (!hasKey) {
                //活动场次id_商品id
                List<String> ids = seckillSessionPojo.getSkuRelationEntities().stream().map(seckillSkuRelationPojo ->
                        seckillSkuRelationPojo.getPromotionSessionId().toString()+"_"+seckillSkuRelationPojo.getSkuId().toString()).collect(Collectors.toList());
                stringRedisTemplate.opsForList().leftPushAll(key, ids);
            }
        });
    }
}
